package org.example.redis;

import org.apache.flink.api.connector.source.ReaderOutput;
import org.apache.flink.api.connector.source.SourceReader;
import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.api.connector.source.lib.util.IteratorSourceReader;
import org.apache.flink.core.io.InputStatus;
import redis.clients.jedis.Jedis;

import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CompletableFuture;

public class RedisSourceReader implements SourceReader<String, RedisSplit> {

    private final Jedis jedis;
    private final String redisKey;
    private final SourceReaderContext context;

    public RedisSourceReader(String host, int port, String redisKey, SourceReaderContext context) {
        this.jedis = new Jedis(host, port);
        this.redisKey = redisKey;
        this.context = context;
    }

    @Override
    public void start() {
        // Start any necessary processes here
    }

    @Override
    public InputStatus pollNext(ReaderOutput<String> output) throws Exception {
        String value = jedis.get(redisKey);
        if (value != null) {
            output.collect(value);
            return InputStatus.MORE_AVAILABLE;
        } else {
            return InputStatus.NOTHING_AVAILABLE;
        }
    }

    @Override
    public List<RedisSplit> snapshotState(long checkpointId) {
        return Collections.emptyList();
    }

    @Override
    public CompletableFuture<Void> isAvailable() {
        return null;
    }

    @Override
    public void addSplits(List<RedisSplit> splits) {

    }

    @Override
    public void notifyNoMoreSplits() {
        // No action needed
    }

    @Override
    public void close() throws Exception {
        jedis.close();
    }
}
